package com.cryann.mqttserver.mqttProcess;

import com.cryann.mqttserver.modle.MqttMessageStore;
import com.cryann.mqttserver.service.Impl.MqttSubjectImpl;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author cRyann
 * @date 2018/12/19 0019 08:24
 * @description:
 */
@Component
@Slf4j
@ProcessType(MqttMessageType.SUBSCRIBE)
public class SubScribe implements BasicProcess<MqttSubscribeMessage> {

    @Autowired
    MqttSubjectImpl mqttSubject;
    @Autowired
    MqttMessageStore mqttMessageStore;

    @Override
    public void process(Channel ctx, MqttSubscribeMessage msg) {
        // TODO 处理
        mqttSubject.addSubject(ctx, msg);

        List<MqttTopicSubscription> list = msg.payload().topicSubscriptions();
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
//        [MQTT-2.3.1-7] SUBACK和UNSUBACK必须包含在对应的SUBSCRIBE和UNSUBSCRIBE报文中使用的报文标识符。
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(msg.variableHeader().messageId());

        int[] integers = list.stream().mapToInt(sub -> sub.qualityOfService().value()).toArray();
        MqttSubAckPayload payload = new MqttSubAckPayload(integers);
        MqttSubAckMessage ackMessage = new MqttSubAckMessage(fixedHeader, variableHeader, payload);
        ctx.writeAndFlush(ackMessage);
        // TODO 获取未发消息，发送消息
//        mqttSubject.publicSub(message);
    }
}
